package com.xiaoa.gossip.cluster.service.impl;

import com.xiaoa.gossip.cluster.command.CurrentNodeState;
import com.xiaoa.gossip.cluster.command.NodeState;
import com.xiaoa.gossip.cluster.command.PingCommand;
import com.xiaoa.gossip.cluster.command.PingNodeState;
import com.xiaoa.gossip.cluster.observer.NodeStateObserver;
import com.xiaoa.gossip.cluster.service.ClusterClientService;
import com.xiaoa.gossip.cluster.service.ClusterService;
import com.xiaoa.gossip.cluster.utils.ClusterUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;
import org.springframework.web.client.RestTemplate;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

import static com.xiaoa.gossip.cluster.command.NodeState.NODE_STATE_FAIL;
import static com.xiaoa.gossip.cluster.command.NodeState.NODE_STATE_P_FAIL;
import static com.xiaoa.gossip.cluster.command.PingCommand.CLUSTERMSG_TYPE_MEET;
import static com.xiaoa.gossip.cluster.command.PingCommand.CLUSTERMSG_TYPE_PONG;
import static com.xiaoa.gossip.cluster.constant.ClusterMethodConstant.PING_METHOD_NAME;

/**
 * @author luokaichuang
 * @time 2020/5/7
 */
@Service
public class ClusterServiceImpl implements ClusterService {

    // 当前节点集合
    private List<PingNodeState> nodeStateList = new Vector<>();

    // map
    private Map<String, PingNodeState> nodeStateMap = new ConcurrentHashMap<>();


    @Autowired
    private RestTemplate restTemplate;

    @Autowired
    private CurrentNodeState currentNodeState;

    @Autowired
    private ClusterClientService clusterClientService;

    // 节点状态监控
    @Autowired(required = false)
    private List<NodeStateObserver> observerList;


    @Value("${gossip.version}")
    private Double version;

    @Value("${gossip.max_ping_fail}")
    private Integer maxPingFail;

    @Value("${gossip.package_node_num}")
    private Integer packageNodeNum;

    @Override
    public PingCommand meet(PingCommand pingCommand) {

        PingNodeState pingNodeState = ClusterUtils.pingCommandToPingNodeState(pingCommand);

        // 创建 pongCommand 对象
        PingCommand pongCommand = createPing();
        pongCommand.setType(CLUSTERMSG_TYPE_PONG);
        pongCommand.setPingNodeStateList(getPartNode(pingNodeState.getId(), false, packageNodeNum));

        // 添加或者更新节点状态
        addOrUpdateNode(pingNodeState, pingNodeState);

        return pongCommand;
    }

    /**
     * 获取部分节点
     *
     * @author jm008783
     * @param: [excludeNodeId] 排除节点id
     * @return: java.util.List<com.xiaoa.gossip.cluster.command.PingNodeState>
     **/
    private synchronized List<PingNodeState> getPartNode(String excludeNodeId, boolean asc, int num) {

        // 先对节点进行排序
        if (asc) {
            ClusterUtils.doSortByTimeAsc(nodeStateList);
        } else {
            ClusterUtils.doSortByTimeDesc(nodeStateList);
        }

        // 挑选出的节点list
        List<PingNodeState> partNodeList = new ArrayList();

        int i = 0;
        for (PingNodeState iNode : nodeStateList) {

            if (i >= num) {
                continue;
            }

            // 忽略掉排除节点
            if (!StringUtils.isEmpty(excludeNodeId) && iNode.getId().equals(excludeNodeId)) {
                continue;
            }

            partNodeList.add(iNode);

            i++;
        }

        return partNodeList;

    }

    @Override
    public PingCommand ping(PingCommand pingCommand) {

        // 创建 pongCommand 对象
        PingCommand pongCommand = createPing();
        pongCommand.setType(CLUSTERMSG_TYPE_PONG);
        pongCommand.setPingNodeStateList(getPartNode(pingCommand.getId(), false, packageNodeNum));

        // 处理ping信息
        doPingHandle(pingCommand);

        return pongCommand;
    }

    @Override
    public PingCommand fail(PingCommand pingCommand) {

        if (nodeStateMap.containsKey(pingCommand.getId())) {

            PingNodeState nodeState = nodeStateMap.get(pingCommand.getId());
            nodeStateList.remove(nodeState);
            nodeStateMap.remove(nodeState.getId());
        }

        return null;
    }

    @Override
    public boolean addToCluster(String nodeIp, int port) {

        PingCommand pingCommand = createPing();

        String url = "http://" + nodeIp + ":" + port + "/cluster/meet";
        PingCommand pongCommand = restTemplate.postForObject(url, pingCommand, PingCommand.class);

        // 处理pong信息
        doPongHandle(pongCommand);

        return true;
    }


    /**
     * 处理ping信息
     *
     * @author jm008783
     * @param: [pingCommand]
     * @return: void
     **/
    private void doPingHandle(PingCommand pingCommand) {
        doPongHandle(pingCommand);
    }

    /**
     * pongCommand 处理器
     *
     * @author jm008783
     * @param: [pongCommand]
     * @return: void
     **/
    @Override
    public synchronized void doPongHandle(final PingCommand pongCommand) {

        PingNodeState nodeState = ClusterUtils.pingCommandToPingNodeState(pongCommand);

        // 处理当前发送过来的节点
        addOrUpdateNode(nodeState, nodeState);

        if (CollectionUtils.isEmpty(pongCommand.getPingNodeStateList())) {
            return;
        }

        // 循环包裹中的节点
        pongCommand.getPingNodeStateList().forEach(n -> addOrUpdateNode(n, nodeState));

    }

    @Override
    public void doPingToNeighbor() {

        int pingNum = Math.max(3, nodeStateList.size() / 3);

        // 包裹节点
        int pgNum = Math.max(3, nodeStateList.size() / 3);

        // 获取最久没有通讯的节点
        List<PingNodeState> pingNodeList = getPartNode(null, true, pingNum);

        // 获取包裹节点
        List<PingNodeState> pgNodeList = getPartNode(null, false, pgNum);

        PingCommand ping = createPing();
        ping.setPingNodeStateList(pgNodeList);

        // 发送ping
        clusterClientService.doBroadcast(PING_METHOD_NAME, pingNodeList, ping);

    }

    @Override
    public List<PingNodeState> noteList() {
        return nodeStateList;
    }

    @Override
    public void doPingFail(PingNodeState pingNodeState) {

        if (nodeStateMap.containsKey(pingNodeState.getId())) {

            PingNodeState nodeState = nodeStateMap.get(pingNodeState.getId());

            // 如果不是下线状态，设置为不联通状态
            if (nodeState.getState() != NODE_STATE_FAIL) {
                nodeState.setState(NODE_STATE_P_FAIL);
            }
        }

    }

    /**
     * 添加或者更新节点状态
     *
     * @param pingCommand 来源节点
     *                    该方法用于更新节点状态，内部  有 observer 可以根据状态改变作出各种处理
     * @author jm008783
     * @param: nodeState 需要更新的节点
     * @return: void
     **/
    @Override
    public synchronized void addOrUpdateNode(PingNodeState nodeState, NodeState pingCommand) {

        // 如果是当前机器，，忽略
        if (isCurrentNote(nodeState)) {
            return;
        }

        PingNodeState oldNode = nodeStateMap.get(nodeState.getId());
        if (oldNode == null) {
            nodeStateList.add(nodeState);
            nodeStateMap.put(nodeState.getId(), nodeState);
        } else {

            // 如果节点比系统中节点新，则更新节点 , 着这是当前节点进行更新
            if (oldNode.getDataVersion() < nodeState.getDataVersion() || isCurrentNote(pingCommand)) {

                PingNodeState oldNodeCopy = new PingNodeState();
                BeanUtils.copyProperties(oldNode, oldNodeCopy);

                // 复制为最新状态
                doUpdateNoteInfo(nodeStateMap.get(nodeState.getId()), nodeState ,  pingCommand );

                // 调用观察者
                for (NodeStateObserver observer : observerList) {
                    observer.doUpdate(oldNodeCopy, nodeState, pingCommand);
                }
            }

        }
    }

    @Override
    public NodeState getNodeState(String id) {
        return nodeStateMap.get(id);
    }

    /**
     *  更新节点最后通讯时间
     * @author jm008783
     * @param: [noteId]
     * @return: void
     **/
    @Override
    public synchronized void updatePingLastTime(String noteId) {
        PingNodeState oldNode = nodeStateMap.get(noteId);
        if(oldNode != null){
            oldNode.setLastPingTime(new Date());
        }
    }

        private boolean isCurrentNote(NodeState pingCommand) {
        return pingCommand.getId().equals(currentNodeState.getId());
    }

    /**
     * 更新节点信息
     *
     * @author jm008783
     * @param: [oldNode, newNote]
     * @return: void
     **/
    private void doUpdateNoteInfo(PingNodeState oldNode, PingNodeState newNote, NodeState pingCommand) {

        // 如果是当前节点，，当前节点目前只能更新其他节点状态
        if (isCurrentNote(pingCommand) && !isCurrentNote(newNote)) {
            oldNode.setState(newNote.getState());
            return;
        }

        oldNode.setSlot(newNote.getSlot());

        // 如果有争议，则使用old 状态
        if (!isArgueStauts(oldNode, newNote)) {
            oldNode.setState(newNote.getState());
        }

        // 如果是需要更新的id 发送来的信息，以发送节点信息为准
        if (oldNode.getId().equals(pingCommand.getId())){
            oldNode.setState(newNote.getState());
        }

        // 更新数据
        oldNode.setMetadata(newNote.getMetadata());
        oldNode.setSlot(newNote.getSlot());
        oldNode.setDataVersion(newNote.getDataVersion());

    }


    /**
     * 判断是否是有争议的状态
     *
     * @author jm008783
     * @param: [nodeState]
     * @return: boolean
     **/
    public boolean isArgueStauts(PingNodeState oldNodeState, PingNodeState newNodeState) {
        if (newNodeState.getState() == NODE_STATE_P_FAIL) {
            return true;
        }

        if (oldNodeState.getState() == NODE_STATE_P_FAIL) {
            return true;
        }

        return false;
    }

    /**
     * 创建ping对象
     *
     * @author jm008783
     * @param: []
     * @return: com.xiaoa.gossip.cluster.command.PingCommand
     **/
    private PingCommand createPing() {

        PingCommand pingCommand = new PingCommand();
        pingCommand.setVer(version);
        pingCommand.setId(currentNodeState.getId());
        pingCommand.setPort(currentNodeState.getPort());
        pingCommand.setCount(0);
        pingCommand.setType(CLUSTERMSG_TYPE_MEET);
        pingCommand.setMyip(currentNodeState.getIp());
        pingCommand.setState(currentNodeState.getState());
        pingCommand.setMyslots(currentNodeState.getSlot());
        pingCommand.setMyMetadata(currentNodeState.getMetadata());
        pingCommand.setDataVersion(currentNodeState.getDataVersion());

        return pingCommand;
    }

}
